package com.link.base.core.basic.interceptors.mybatis.system;

import com.alibaba.fastjson.JSONObject;
import com.link.base.base.trackdisplay.dao.mybatis.mapper.ReturnServiceMapper;
import com.link.base.base.trackdisplay.model.TrackDisplay;
import com.link.base.base.trackdisplay.service.TrackDisplayService;
import com.link.base.base.user.model.User;
import com.link.base.basic.service.LovService;
import com.link.core.basic.interceptor.SecurityInterceptor;
import com.link.core.cllog.LCLogger;
import com.link.core.kafka.KafkaConnectCondition;
import com.link.core.kafka.conf.KafkaEnv;
import com.link.core.util.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.context.annotation.Conditional;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;

/**
 * @Author: qinxuegang
 * @Description:
 * @Date: 2018/12/24 14:46
 */
@Component
@Conditional(KafkaConnectCondition.class)
public class KafkaConsumer {
    @Resource
    private TrackDisplayService trackDisplayService;
    @Resource
    private LovService lovService;
    @Resource
    private ReturnServiceMapper returnServiceMapper;

    private static String TOPIC_NAME = KafkaEnv.getTrackTopic();

    private static final Logger logger = LogManager.getLogger(KafkaConsumer.class);


    /**
     * 监听CDC_CONSUMER_RST主题,有消息就读取
     *
     * @param
     */
    @KafkaListener(topics = "KT_FIELD_TRACK")
    public void receiveMessage(List<ConsumerRecord> records) {
        try {
            logger.info("监听到KT_FIELD_TRACK！！");
            for (ConsumerRecord record : records) {
                try {
                    if (record.value() != null) {
                        TrackDisplay trackDisplay = JSONObject.toJavaObject(JSONObject.parseObject(record.value().toString()), TrackDisplay.class);
                        User user = new User();
                        user.setId(trackDisplay.getAttr77());
                        user.setCorpid(trackDisplay.getCorpid());
                        user.setPostnId(trackDisplay.getPostnId());
                        user.setOrgId(trackDisplay.getOrgId());
                        SecurityInterceptor.tempUser.set(user);
                        if ("lovValue".equals(trackDisplay.getAttr1())) {
                            String oldLovValue = "";
                            if (StringUtils.isNotBlank(trackDisplay.getUpdBeforeInfo())) {
                                oldLovValue = lovService.getLovName(trackDisplay.getAttr2(), trackDisplay.getUpdBeforeInfo());
                            }
                            String newLovValue = "";
                            if (StringUtils.isNotBlank(trackDisplay.getUpdInfo())) {
                                newLovValue = lovService.getLovName(trackDisplay.getAttr2(), trackDisplay.getUpdInfo());
                            }
                            trackDisplay.setDisplayBeforeInfo(oldLovValue);
                            trackDisplay.setDisplayInfo(newLovValue);
                        } else if ("CorrespondingValue".equals(trackDisplay.getAttr1())) {
                            String tempParam = trackDisplay.getAttr3();
                            if (StringUtils.isNotBlank(trackDisplay.getUpdBeforeInfo())) {
                                String sqlOld = tempParam + " = '" + trackDisplay.getUpdBeforeInfo() + "'";
                                trackDisplay.setAttr5(sqlOld);
                                Map<String, Object> oldMap = returnServiceMapper.returnService(trackDisplay);
                                if (oldMap != null) {
                                    trackDisplay.setDisplayBeforeInfo((String) oldMap.get(trackDisplay.getAttr4()));
                                }
                            }
                            if (StringUtils.isNotBlank(trackDisplay.getUpdInfo())) {
                                String sqlNew = tempParam + " = '" + trackDisplay.getUpdInfo() + "'";
                                trackDisplay.setAttr5(sqlNew);
                                Map<String, Object> newMap = returnServiceMapper.returnService(trackDisplay);
                                if (newMap != null) {
                                    trackDisplay.setDisplayInfo((String) newMap.get(trackDisplay.getAttr4()));
                                }
                            }
                        } else if ("realValue".equals(trackDisplay.getAttr1())) {
                            trackDisplay.setDisplayBeforeInfo(trackDisplay.getUpdBeforeInfo());
                            trackDisplay.setDisplayInfo(trackDisplay.getUpdInfo());
                        }
                        trackDisplayService.insert(trackDisplay);
                    }
                } catch (Exception e) {
                    //避免因为某个字段报错而影响其他项目报错
                    LCLogger.withException(e);
                }
            }

        } catch (Exception e) {
            LCLogger.withException(e);
        }
    }
}
